package cn.doitedu.api;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Date;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/2/28
 * @Desc: 学大数据，上多易教育
 *
 *     connect算子演示
 **/
public class _29_Connect_Demo {

    public static void main(String[] args) throws Exception {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        // o1,2024-02-28 11:22:30.200
        DataStreamSource<String> stream1 = env.socketTextStream("doitedu", 8891);


        // {"order_id":"o1","create_time":1709091150000}
        DataStream<JSONObject> stream2 = env.socketTextStream("doitedu", 8892).map(s-> JSON.parseObject(s));

        // 通过connect，来对上面两个进行连接，结果成为统一类型的一个流
        DataStream<GlobalBean> resStream = stream1.connect(stream2)
                .map(new CoMapFunction<String, JSONObject, GlobalBean>() {

                    @Override
                    public GlobalBean map1(String csvLine) throws Exception {
                        String[] split = csvLine.split(",");
                        // o1,2024-02-28 11:22:30.200
                        Date date = DateUtils.parseDate(split[1], "yyyy-MM-dd HH:mm:ss.SSS");
                        long time = date.getTime();

                        return new GlobalBean(split[0], time);
                    }

                    @Override
                    public GlobalBean map2(JSONObject jsonObject) throws Exception {

                        //  {"order_id":"o1","create_time":1632873459874}

                        return new GlobalBean(jsonObject.getString("order_id"), jsonObject.getLongValue("create_time"));
                    }
                });

        resStream.print();

        env.execute();

    }
    
}


@Data
@NoArgsConstructor
@AllArgsConstructor
class GlobalBean {
    private String orderId;
    private long createTime;

}
